<?php
namespace Weibo\Lib;
use \AMQPConnection;
use \AMQPChannel;
use \AMQPExchange;
use \AMQPQueue;
use \Exception;

class MqLib {
	
	private static $_rabbit;
	
	public static function connet(){
		if (!self::$_rabbit) {
			$config = C('MQ');
			self::$_rabbit = new RabbitMQLib($config);
		}
		
		return self::$_rabbit;
	}
	
	public static function add($data = '', $queue = ''){
		if(!$data || !$queue){
			return FALSE;
		} else {
			$rabbit = self::connet();
			if ($rabbit) {
				$rabbit -> exchange_name = $queue.'-exchange';
				$rabbit -> queue_name = $queue.'-queue';
				$rabbit -> route_key = $queue.'-route';
				
				if (is_array($data)) {
					foreach ($data as $item){
						$rabbit -> send($item);
					}
					return TRUE;
				} else {
					return $rabbit -> send($data);
				}
			} else {
				return FALSE;
			}
		}
	}
}

class RabbitMQLib{

	public $durable = TRUE;			//持久化,默认TRUE
	public $autodelete = FALSE;		//自动删除exchange is deleted when all queues have finished using it. queue is deleted when last consumer unsubscribes
	public $mirror = FALSE;			//镜像队列,打开后消息会在节点之间复制,有master和slave的概念

	private $configs = array();		//连接配置
	
	private $exchange_name = '';	//交换机名称
	private $queue_name = '';		//队列名称
	private $route_key = '';		//路由名称
	private $eqr_changed = FALSE;	//改变exchange,queue,route设置是将其设为TRUE,以便重新配置AMQP连接
	
	private $_conn = NULL;
	private $_exchange = NULL;
	private $_channel = NULL;
	private $_queue = NULL;

	/**
	 * 构造方法
	 * @param	array	$configs			连接配置 array('host'=>$host,'port'=>5672,'username'=>$username,'password'=>$password,'vhost'=>'/')
	 * @param	string	$exchange_name
	 * @param	string	$queue_name
	 * @param	string	$route_key
	 */
	public function __construct($configs = array(), $exchange_name = '', $queue_name = '', $route_key = '') {
		$this -> setConfigs($configs);
		$this -> exchange_name = $exchange_name;
		$this -> queue_name = $queue_name;
		$this -> route_key = $route_key;
	}
	
	public function __set($name, $value){
		if (in_array($name, array('exchange_name', 'queue_name', 'route_key'))) {
			$this -> $name = $value;
			$this -> eqr_changed || $this -> eqr_changed = TRUE;
		}
	}

	private function setConfigs($configs) {
		if (!is_array($configs)) {
			throw new Exception('configs is not array');
		}
		if (!($configs['host'] && $configs['port'] && $configs['username'] && $configs['password'])) {
			throw new Exception('configs is empty');
		}
		if (empty($configs['vhost'])) {
			$configs['vhost'] = '/';
		}
		$configs['login'] = $configs['username'];
		unset($configs['username']);
		$this -> configs = $configs;
	}

	/**
	 * 连接到AMQP
	 * @throws Exception
	 */
	private function open() {
		if (!$this -> _conn) {
			try {
				$this -> _conn = new AMQPConnection($this -> configs);
				$this -> _conn -> connect();
				$this -> initConnection();
			} catch (AMQPConnectionException $ex) {
				throw new Exception('cannot connection rabbitmq',500);
			}
		} elseif ((!$this -> _exchange || !$this -> _channel || !$this -> _queue) && $this -> eqr_changed) {
			try {
				$this -> initConnection();
			} catch (AMQPConnectionException $ex) {
				throw new Exception('cannot connection rabbitmq',500);
			}
		}
	}
	
	/**
	 * 初始化连接和配置
	 * @throws Exception
	 */
	private function initConnection() {
		if (empty($this -> exchange_name) || empty($this -> queue_name) || empty($this -> route_key)) {
			throw new Exception('rabbitmq exchange_name or queue_name or route_key is empty',500);
		}
		$this -> _channel = new AMQPChannel($this -> _conn);
		$this -> _exchange = new AMQPExchange($this -> _channel);
		$this -> _exchange -> setName($this -> exchange_name);

		$this -> _exchange -> setType(AMQP_EX_TYPE_DIRECT);
		if ($this -> durable){
			$this -> _exchange -> setFlags(AMQP_DURABLE);
		}
		if ($this -> autodelete){
			$this -> _exchange -> setFlags(AMQP_AUTODELETE);
		}
		$this -> _exchange -> declare();

		$this -> _queue = new AMQPQueue($this -> _channel);
		$this -> _queue -> setName($this -> queue_name);
		if ($this -> durable){
			$this -> _queue -> setFlags(AMQP_DURABLE);
		}
		if ($this -> autodelete){
			$this -> _queue -> setFlags(AMQP_AUTODELETE);
		}
		if ($this -> mirror){
			$this -> _queue -> setArgument('x-ha-policy', 'all');
		}
			
		$this -> _queue -> declare();
		$this -> _queue -> bind($this -> exchange_name, $this -> route_key);
	}


	/**
	 * 生产者发送消息
	 * @param	string	$msg	消息内容
	 */
	public function send($msg) {
		try {
			$this -> open();
		} catch (Exception $e) {
			//TODO 这里做日志记录
			$e -> getMessage();
			
			return FALSE;
		}
		
		if(is_array($msg)){
			$msg = json_encode($msg);
		}else{
			$msg = trim(strval($msg));
		}
		return $this -> _exchange -> publish($msg, $this -> route_key);
	}
	
	/**
	 * 消费者调用
	 *
	 * function processMessage($envelope, $queue) {
	 * 		$msg = $envelope -> getBody();
	 * 		echo $msg."\n";									//处理消息
	 * 		$queue -> ack($envelope -> getDeliveryTag());	//手动应答
	 * }
	 *
	 * @param	callable	$fun_name		消息处理回调函数
	 * @param	boolean		$autoack		自动应答,告诉RabbitMQ我已处理完毕
	 * @return	boolean
	 */
	public function run($fun_name, $autoack = TRUE){
		$this -> open();
		if (!$fun_name || !$this -> _queue) return FALSE;
		while(TRUE){
			if ($autoack) $this -> _queue -> consume($fun_name, AMQP_AUTOACK);
			else $this -> _queue -> consume($fun_name);
		}
	}

	/**
	 * 断开AMQP连接
	 */
	public function close() {
		if ($this -> _conn) {
			$this -> _conn -> disconnect();
		}
	}
	
	public function __sleep() {
		$this -> close();
		return array_keys(get_object_vars($this));
	}

	public function __destruct() {
		$this -> close();
	}

}